package com.github.yoojia.mqtt;


import com.parkingwang.lang.Required;

import java.io.IOException;

/**
 * @author Yoojia Chen (yoojiachen@gmail.com)
 * @since 1.0.0
 */
class RealCall implements Call {

    private final Subscriber mReplySubscriber;
    private final Message mRequestMessage;
    private final MQTTSocket mSocket;
    private final MatchedId mMatchedId;

    public RealCall(Subscriber subscriber, Message requestMessage, MQTTSocket socket) {
        mReplySubscriber = subscriber;
        mRequestMessage = requestMessage;
        mSocket = socket;
        mMatchedId = new MatchedId(mRequestMessage.messageId);
    }

    @Override
    public Message execute() throws Exception {
        final Required<Message> value = new Required<>();
        final FinalLong id = new FinalLong();
        final long sid = mReplySubscriber.addHandler(mMatchedId, new MessageHandler() {
            @Override
            public void onMessage(MQTTSocket socket, Message reply) {
                mReplySubscriber.removeHandler(id.get());
                value.set(reply);
            }
        });
        id.set(sid);
        try {
            mSocket.sendThrows(mRequestMessage);
            return value.getAwaitChecked();
        } catch (IOException ioe) {
            mSocket.removeReplyMessageHandler(id.get());
            throw ioe;
        }
    }

    @Override
    public void async(final MessageCallback callback) {
        final FinalLong id = new FinalLong();
        final long sid = mReplySubscriber.addHandler(mMatchedId, new MessageHandler() {
            @Override
            public void onMessage(MQTTSocket socket, Message reply) {
                mReplySubscriber.removeHandler(id.get());
                try {
                    callback.onMessage(socket, reply);
                } catch (Exception e) {
                    callback.onError(e);
                }
            }
        });
        id.set(sid);
        try {
            mSocket.sendThrows(mRequestMessage);
        } catch (IOException ioe) {
            mSocket.removeReplyMessageHandler(id.get());
            callback.onError(ioe);
        }
    }

    private static class FinalLong {
        private long mValue;

        void set(long value) {
            mValue = value;
        }

        long get() {
            return mValue;
        }
    }

    private static class MatchedId implements LocalFilter {

        private final long mExpectId;

        public MatchedId(long expectId) {
            mExpectId = expectId;
        }

        @Override
        public boolean test(Message item) {
            return mExpectId == item.messageId;
        }
    }
}
